package cn.edu.gmall.realtime.app.func;

import cn.edu.gmall.realtime.utils.DimUtil;
import cn.edu.gmall.realtime.utils.ThreadPoolUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class DimJoinAsyncFunction<T> extends RichAsyncFunction<T, T> {
    //声明线程池对象
    private ThreadPoolExecutor threadPoolExecutor;
    //声明属性
    private final String tableName;

    public DimJoinAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化线程池
        threadPoolExecutor = ThreadPoolUtil.getThreadPool();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                //0.获取查询条件
                String key = getKey(input);
                //1.查询维度信息
                JSONObject dimInfo = DimUtil.getDimInfo(tableName, key);
                //2.关联到事实数据上
                if (dimInfo != null && dimInfo.size() > 0) {
                    try {
                        join(input, dimInfo);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                //3.继续向下游传输
                resultFuture.complete(Collections.singletonList(input));
            }
        };
        threadPoolExecutor.submit(runnable);
    }

    //获取数据中的所要关联维度的主键
    public abstract String getKey(T input);

    //关联事实数据和维度数据
    public abstract void join(T input, JSONObject dimInfo) throws Exception;
}
